package com.vcc.web.core.component;

import com.vcc.core.exception.ProgramException;
import com.vcc.core.util.JsonUtil;
import com.vcc.data.entity.Wallet;
import com.vcc.service.client.vo.wellat.WalletVo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.MediaType;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import java.util.stream.Collectors;

@Component
public class SseComponent {

    private static Map<String, SseEmitter> sseCache = new ConcurrentHashMap<>();

    private static final Logger logger = LoggerFactory.getLogger(SseComponent.class);

    public SseEmitter createSseConnect(String clientId, Long userId) {
        // 设置超时时间，0表示不过期。默认30秒，超过时间未完成会抛出异常：AsyncRequestTimeoutException
        SseEmitter sseEmitter = new SseEmitter(0L);
        // 注册回调
        String id = userId + "-" + clientId;
        sseEmitter.onCompletion(completionCallBack(id));
        sseCache.put(id, sseEmitter);
        logger.info("创建新的sse连接，当前用户：{}", id);
        try {
            sseEmitter.send(SseEmitter.event().data(id));
        } catch (IOException e) {
            logger.error("SseEmitterServiceImpl[createSseConnect]: 创建长链接异常，客户端ID:{}", id, e);
            throw new ProgramException("创建连接异常！");
        }
        return sseEmitter;
    }

    public void closeSseConnect(String clientId, Long userId) {
        String id = userId + "-" + clientId;
        SseEmitter sseEmitter = sseCache.get(id);
        if (sseEmitter != null) {
            removeUser(id);
            sseEmitter.complete();
        }
    }

    // 根据客户端id获取SseEmitter对象
    public SseEmitter getSseEmitterByClientId(String id) {
        return sseCache.get(id);
    }

    // 推送消息到客户端，此处结合业务代码，业务中需要推送消息处调用即可向客户端主动推送消息
    @Async
    public void sendRefreshUsableQuotaMessage(Long userId,List<Wallet> wallets) {
        Map<String, Object> map = new HashMap<>();
        map.put("type", "refreshUsableQuota");
        map.put("data", wallets.stream().map(WalletVo::new).collect(Collectors.toList()));
        for (Map.Entry<String, SseEmitter> entry : sseCache.entrySet()) {
            if (entry.getKey().startsWith(userId.toString())) {
                //断开的舍弃
                if (entry.getValue() == null) {
                    removeUser(entry.getKey());
                } else {
                    sendMsgToClientByClientId(entry.getValue(), entry.getKey(), map);
                }
            }
        }
    }


    private void sendMsgToClientByClientId(SseEmitter sseEmitter, String key, Map<String, Object> data) {
        SseEmitter.SseEventBuilder sendData = SseEmitter.event().data(JsonUtil.toJson(data), MediaType.APPLICATION_JSON);
        try {
            sseEmitter.send(sendData);
        } catch (IOException e) {
            // 推送消息失败，记录错误日志，进行重推
            logger.error("SseEmitterServiceImpl[sendMsgToClient]: 推送消息失败：{},尝试进行重推", JsonUtil.toJson(data), e);
            boolean isSuccess = true;
            // 推送消息失败后，每隔1s推送一次，推送3次
            for (int i = 1; i <= 3; i++) {
                try {
                    Thread.sleep(1000);
                    sseEmitter.send(sendData);
                } catch (Exception ex) {
                    logger.error("SseEmitterServiceImpl[sendMsgToClient]：的第{}次消息重推失败", i, ex);
                    if (i == 3) {
                        removeUser(key);
                    }
                    continue;
                }
                logger.info("SseEmitterServiceImpl[sendMsgToClient]：的第{}次消息重推成功,{}", i, JsonUtil.toJson(data));
                return;
            }
        }
    }

    /**
     * 长链接完成后回调接口(即关闭连接时调用)
     *
     * @param clientId 客户端ID
     * @return java.lang.Runnable
     * @author re
     * @date 2021/12/14
     **/
    private Runnable completionCallBack(String clientId) {
        return () -> {
            logger.info("结束连接：{}", clientId);
            removeUser(clientId);
        };
    }

    /**
     * 连接超时时调用
     *
     * @param clientId 客户端ID
     * @return java.lang.Runnable
     * @author re
     * @date 2021/12/14
     **/
    private Runnable timeoutCallBack(String clientId) {
        return () -> {
            logger.info("连接超时：{}", clientId);
            removeUser(clientId);
        };
    }

    /**
     * 推送消息异常时，回调方法
     *
     * @param clientId 客户端ID
     * @return java.util.function.Consumer<java.lang.Throwable>
     * @author re
     * @date 2021/12/14
     **/
    private Consumer<Throwable> errorCallBack(String clientId) {
        return throwable -> {
            logger.error("SseEmitterServiceImpl[errorCallBack]：连接异常,客户端ID:{}", clientId);

            // 推送消息失败后，每隔10s推送一次，推送5次
            for (int i = 0; i < 5; i++) {
                try {
                    Thread.sleep(10000);
                    SseEmitter sseEmitter = sseCache.get(clientId);
                    if (sseEmitter == null) {
                        logger.error("SseEmitterServiceImpl[errorCallBack]：第{}次消息重推失败,未获取到 {} 对应的长链接", i + 1, clientId);
                        continue;
                    }
                    sseEmitter.send("失败后重新推送");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };
    }

    /**
     * 移除用户连接
     *
     * @param clientId 客户端ID
     * @author re
     * @date 2021/12/14
     **/
    private void removeUser(String clientId) {
        sseCache.remove(clientId);
        logger.info("SseEmitterServiceImpl[removeUser]:移除用户：{}", clientId);
    }


    @Scheduled(cron = "0 0/10 * * * ?")
    public void heartbeat() {
        logger.info("心跳机制触发");
        Map<String, Object> map = new HashMap<>();
        map.put("type", "ping");
        for (Map.Entry<String, SseEmitter> entry : sseCache.entrySet()) {
            //断开的舍弃
            if (entry.getValue() == null) {
                removeUser(entry.getKey());
            } else {
                sendMsgToClientByClientId(entry.getValue(), entry.getKey(), map);
            }
        }
    }
}
